package com.motu.vertx.module.utility.redis;

import com.motu.vertx.module.utility.base.ServerConfig;
import com.motu.vertx.module.utility.toolset.Tool;
import io.vertx.core.AsyncResult;
import io.vertx.core.CompositeFuture;
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.redis.client.Redis;
import io.vertx.redis.client.RedisAPI;
import io.vertx.redis.client.RedisConnection;
import io.vertx.redis.client.RedisOptions;
import io.vertx.redis.client.Response;
import io.vertx.redis.client.ResponseType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Stream;

public class RedisTools {

    private static final Logger logger = LoggerFactory.getLogger("RedisTools");
    private static final int MAX_RECONNECT_REDIS_RETRIES = 16;// 尝试重连redis最大次数
    private static final int MAX_WAITING_HANDLERS = 1000000;// redis请求队列最大长度，默认值是2048，如果redis请求数超过这个值会请求失败

    private static RedisAPI client;// redis连接对象

    /***
     * web后台用户token信息，包含token和tokentime以及rolelist
     * @param userName
     * @return
     */
    public static final String getWebUserTokenKey(String userName,String token){
        return "webuser."+userName+"."+token;
    }

    // 设置redis链接
    public static void setRedisAPI(RedisAPI redisAPI) {
        client = redisAPI;
    }

    /**
     * 初始化redis连接（调用这个方法前提是需要先初始化服务器配置信息）
     * @param vertx
     * @return
     */
    public static Future<Void> init(Vertx vertx) {
        return init(vertx, ServerConfig.getAppConfig());
    }

    /**
     * 初始化redis连接
     * @param vertx
     * @param appConfig redis配置信息
     * @return
     */
    public static Future<Void> init(Vertx vertx, JsonObject appConfig) {
        Promise promise = Promise.promise();
        createRedisClient(vertx, appConfig, reconnectionCallback -> {
            // redis重连成功给redisClient重新赋值
            client = reconnectionCallback.result();
            logger.info("redis reconnection callback update redisClient");
        }).onComplete(redisHandler -> {
            if (redisHandler.succeeded()) {
                client = redisHandler.result();
                promise.complete();
            } else {
                logger.error("createRedisClient failed cause:{}", Tool.getException(redisHandler.cause()));
                promise.fail(redisHandler.cause());
            }
        });
        return promise.future();
    }

    /**
     * 创建redis连接（传递回调方法，用于断线重连后更新redis连接对象）
     */
    public static Future<RedisAPI> createRedisClient(Vertx vertx, JsonObject appConfig, Handler<AsyncResult<RedisAPI>> callback) {
        RedisOptions redisOptions = new RedisOptions();
        redisOptions.setMaxWaitingHandlers(MAX_WAITING_HANDLERS);
        JsonArray redisConnectionArray = appConfig.getJsonArray("redis-connection");
        for (int i = 0; i < redisConnectionArray.size(); i++) {
            String conStr = redisConnectionArray.getString(i);
            logger.info("###################### redis connection config:" + conStr);
            redisOptions.addConnectionString(conStr);
        }
        Promise<RedisAPI> promise = Promise.promise();
        Redis.createClient(vertx, redisOptions)
                .connect(onConnect -> {
                    if (onConnect.succeeded()) {
                        RedisConnection client = onConnect.result();
                        logger.info("###################### redis connection success");

                        /**
                         * 确保客户端在错误时重新连接
                         * 例如，底层TCP连接已关闭，但客户端还不知道
                         * 客户端尝试使用断开的连接与服务器通信。将引发一个异常
                         */
                        client.exceptionHandler(handler -> {// redis连接发生异常的时候执行
                            logger.error("###################### redis connection exception");
                            attemptReconnectRedis(vertx, appConfig, 0, callback);// 尝试重连redis
                        });

                        /**
                         * 确保客户端在连接关闭时重新连接
                         * 例如，底层TCP连接通过正常的4路握手关闭
                         * 此处理程序将立即得到通知
                         */
                        client.endHandler(handler -> {// redis连接被关闭的时候执行
                            logger.error("###################### redis connection end");
                            attemptReconnectRedis(vertx, appConfig, 0, callback);// 尝试重连redis
                        });

                        promise.complete(RedisAPI.api(client));
                    } else {
                        logger.error("###################### redis connection error cause = " + Tool.getException(onConnect.cause()));
                        promise.fail(onConnect.cause());
                    }
                });
        return promise.future();
    }

    /**
     * 尝试重连redis
     *
     * @param retry 重连次数
     */
    private static void attemptReconnectRedis(Vertx vertx, JsonObject appConfig, int retry, Handler<AsyncResult<RedisAPI>> callback) {
        if (retry > MAX_RECONNECT_REDIS_RETRIES) {
            logger.error("###################### retry reconnection redis max");
        } else {
            logger.info("###################### retry reconnection redis num = " + retry);
            //10,20,40,80,160,320,640,1280,2560,5120,10240,10240,10240,10240,10240,10240
            long backoff = (long) (Math.pow(2, Math.min(retry, 10)) * 10);// 重连时间，最大10240ms
            vertx.setTimer(backoff, timer -> {
                createRedisClient(vertx, appConfig, callback).onComplete(handler -> {
                    if (handler.failed()) {
                        attemptReconnectRedis(vertx, appConfig, retry + 1, callback);
                    } else {
                        // 重连成功需要执行回调方法
                        if (callback != null) {
                            callback.handle(Future.succeededFuture(handler.result()));
                        }
                    }
                });
            });
        }
    }

    @SuppressWarnings("unchecked")
    private static List<?> toPayload(Object... parameters) {
        List<Object> result = new ArrayList<>(parameters.length);

        for (Object param : parameters) {
            // unwrap
            if (param instanceof JsonArray) {
                param = ((JsonArray) param).getList();
            }
            // unwrap
            if (param instanceof JsonObject) {
                param = ((JsonObject) param).getMap();
            }

            if (param instanceof Collection) {
                ((Collection<?>) param).stream().filter(Objects::nonNull).forEach(result::add);
            } else if (param instanceof Map) {
                for (Map.Entry<?, ?> pair : ((Map<?, ?>) param).entrySet()) {
                    result.add(pair.getKey());
                    result.add(pair.getValue().toString());
                }
            } else if (param instanceof Stream) {
                ((Stream<?>) param).forEach(e -> {
                    if (e instanceof Object[]) {
                        Collections.addAll(result, (Object[]) e);
                    } else {
                        result.add(e);
                    }
                });
            } else if (param instanceof Buffer) {
                result.add(((Buffer) param).getBytes());
            } else if (param != null) {
                result.add(param.toString());
            }
        }
        return result;
    }

    private static JsonObject toJsonObject(Response response) {
        final JsonObject json = new JsonObject();
        int size = response.size();
        if (size == 0) {
            return json;
        }
        if (size % 2 != 0) {
            logger.error("Number of key is not even");
            return json;
        }

		/**
		 * redis3.x和redis6.x版本返回的数据格式不同，需要兼容
		 * redis3.x:[key1,value1,key2,value2]
		 * redis6.x:{key1:value1,key2:value2} (不是json格式)
		 */

		if (response.isMap()){// redis6.x
			for (Response item : response) {
				String key = item.get(0).toString();
				String value = item.get(1).toString();
				json.put(key, value);
			}
		} else {// redis3.x response.isArray()
			for (int i = 0; i < size; i = i + 2) {
				String key = response.get(i).toString();
				Response value = response.get(i + 1);
				json.put(key, value.toString());
			}
		}

        return json;
    }

    private static JsonObject toJsonObject(JsonArray fields, Response response) {
        final JsonObject json = new JsonObject();
        if (response.size() == 0) {
            return json;
        }
        for (int i = 0; i < fields.size(); i++) {
            String key = fields.getString(i);
            Response value = response.get(i);
            if (value == null) {
                continue;
            }
			json.put(key, value.toString());
        }
        return json;
    }

    private static JsonArray toJsonArray(Response response) {
        final JsonArray json = new JsonArray();

        if (response.type() != ResponseType.MULTI) {
            switch (response.type()) {
                case NUMBER:
                    json.add(response.toNumber());
                    break;
                case SIMPLE:
                case BULK:
                    json.add(response.toString());
                    break;
            }
            return json;
        }

        for (Response r : response) {
            if (r == null) {
                json.addNull();
            } else {
                switch (r.type()) {
                    case NUMBER:
                        json.add(r.toNumber());
                        break;
                    case SIMPLE:
                    case BULK:
                        json.add(r.toString());
                        break;
                    case MULTI:
						// redis6.x获取带分数的排行榜，返回的就是MULTI类型
						for (Response r1 : r) {
							json.add(r1.toString());
						}
                        break;
                }
            }
        }

        return json;
    }

    /***
     * 获取表自增值的hash key
     * @return
     */
    public static final String getTableidKey() {
        return "tableid";
    }

    public static void set(String key, String value, Handler<AsyncResult<Void>> onComplete) {
        List<String> list = (List<String>) toPayload(key, value);
        client.set(list, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture());
            } else {
                logger.error("redis error set key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    public static Future<Void> set(String key, String value) {
        Promise<Void> promise = Promise.promise();
        set(key, value, promise);
        return promise.future();
    }

    public static void get(String key, Handler<AsyncResult<String>> onComplete) {
        client.get(key, res -> {
            if (res.succeeded()) {
                Response response = res.result();
                if (response == null) {
                    onComplete.handle(Future.succeededFuture(""));
                } else {
                    onComplete.handle(Future.succeededFuture(response.toString()));
                }
            } else {
                logger.error("redis error get key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    public static Future<String> get(String key) {
        Promise<String> promise = Promise.promise();
        get(key, promise);
        return promise.future();
    }

    // 给指定key加上增量值，如果key不存在则会初始化为0后再增加，返回增加后的结果
    public static void add(String key, long value, Handler<AsyncResult<Long>> onComplete) {
        client.incrby(key, String.valueOf(value), res -> {
            if (res.succeeded()) {
                Response response = res.result();
                if (response == null) {
                    onComplete.handle(Future.succeededFuture(0l));
                } else {
                    onComplete.handle(Future.succeededFuture(response.toLong()));
                }
            } else {
                logger.error("redis error add key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

	public static Future<Long> add(String key, long value) {
		Promise<Long> promise = Promise.promise();
		add(key, value, promise);
		return promise.future();
	}

    // 给指定key扣除增量值，如果key不存在则会初始化为0后再扣除，返回扣除后的结果
    public static void sub( String key, long value, Handler<AsyncResult<Long>> onComplete) {
        client.decrby(key, String.valueOf(value), res -> {
            if (res.succeeded()) {
                Response response = res.result();
                if (response == null) {
                    onComplete.handle(Future.succeededFuture(0l));
                } else {
                    onComplete.handle(Future.succeededFuture(response.toLong()));
                }
            } else {
                logger.error("redis error sub key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

	public static Future<Long> sub(String key, long value) {
		Promise<Long> promise = Promise.promise();
		sub(key, value, promise);
		return promise.future();
	}

    @SuppressWarnings("unchecked")
    public static Future<Boolean> setHashJsonObject( String key, JsonObject obj) {
        Promise<Boolean> promise = Promise.promise();
        setHashJsonObject(key, obj, promise);
        return promise.future();
    }

    @SuppressWarnings("unchecked")
    public static void setHashJsonObject( String key, JsonObject obj, Handler<AsyncResult<Boolean>> onComplete) {
        List<String> list = (List<String>) toPayload(key, obj);
        client.hmset(list, res -> {
            if (res.failed()) {
                logger.error("redis error setHashJsonObject key:{} value:{} exception:{}", key, obj, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            } else {
                onComplete.handle(Future.succeededFuture(true));
            }
        });
    }

    public static void getHashJsonObject( String key, Handler<AsyncResult<JsonObject>> onComplete) {
        client.hgetall(key, res -> {
            if (res.succeeded()) {
                JsonObject object = toJsonObject(res.result());
                onComplete.handle(Future.succeededFuture(object));
            } else {
                logger.error("redis error getHashJsonObject key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    public static Future<JsonObject> getHashJsonObject( String key) {
        Promise<JsonObject> promise = Promise.promise();
        getHashJsonObject(key, promise);
        return promise.future();
    }

    public static void getHashJsonObject( String key, Handler<AsyncResult<JsonObject>> onComplete, String... fields) {
        if (fields.length == 0) {
            getHashJsonObject( key, onComplete);
        } else {
            JsonArray fieldArray = new JsonArray();
            fieldArray.add("id");//默认需要有一个id字段，防止查询的字段为空导致查询结果为空
            for (int i = 0; i < fields.length; i++) {
                fieldArray.add(fields[i]);
            }
            getHashJsonObject( key, fieldArray, onComplete);
        }
    }

    public static Future<JsonObject> getHashJsonObject( String key, String... fields) {
        Promise<JsonObject> promise = Promise.promise();
        if (fields.length == 0) {
           getHashJsonObject( key, promise);
        } else {
            JsonArray fieldArray = new JsonArray();
            fieldArray.add("id");//默认需要有一个id字段，防止查询的字段为空导致查询结果为空
            for (int i = 0; i < fields.length; i++) {
                fieldArray.add(fields[i]);
            }
            getHashJsonObject( key, fieldArray, promise);
        }
        return promise.future();
    }

    public static void getHashJsonObject( String key, JsonArray fields, Handler<AsyncResult<JsonObject>> onComplete) {
        // 字段列表不能为空
        if (fields == null || fields.size() == 0) {
            onComplete.handle(Future.succeededFuture(new JsonObject()));
            return;
        }

        // 字段列表必须是String列表，如果不是则转换成String类型
        if (!(fields.getValue(0) instanceof String)) {
            JsonArray strFields = new JsonArray();
            for (int i = 0; i < fields.size(); i++) {
                strFields.add(String.valueOf(fields.getValue(i)));
            }
            fields = strFields;
        }

        if (fields.size() == 0) {
            logger.error("getHashJsonObject error key:{} fields:{} cause:{}", key, fields, "fields size == 0");
            onComplete.handle(Future.succeededFuture(new JsonObject()));
            return;
        }

        List<String> args = new ArrayList<>();
        args.add(key);
        args.addAll(fields.getList());
        JsonArray fields2 = fields;
        client.hmget(args, res -> {
            if (res.succeeded()) {
                JsonObject object = toJsonObject(fields2, res.result());
                onComplete.handle(Future.succeededFuture(object));
            } else {
                logger.error("redis error getHashJsonObject key:{} fields:{} exception:{}", key, fields2, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    public static Future<JsonObject> getHashJsonObject( String key, JsonArray fields) {
        Promise<JsonObject> promise = Promise.promise();
        getHashJsonObject( key, fields, promise);
        return promise.future();
    }

    public static void incrTableId( String key, String fieldName, Handler<AsyncResult<Long>> onComplete) {
        client.hincrby(key, fieldName, "1", res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toLong()));
            } else {
                logger.error("redis error incrTableId key:{} fieldName:{} exception:{}", key, fieldName, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /***
     * 获取表的自增id，并且+1
     * @param tableName 表名
     * @param onComplete
     */
    public static void incrTableId( String tableName, Handler<AsyncResult<Long>> onComplete) {
        client.hincrby(getTableidKey(), tableName, "1", res -> {
            if (res.failed()) {
                logger.error("redis error incrTableId tableName:{} exception:{}", tableName, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            } else {
                onComplete.handle(Future.succeededFuture(res.result().toLong()));
            }
        });

    }

    /**
     * 获取表的自增id，并且+1
     * @param tableName 表名
     */
    public static Future<Long> incrTableId( String tableName) {
        Promise<Long> promise = Promise.promise();
        incrTableId( tableName, promise);
        return promise.future();
    }

    @SuppressWarnings("unchecked")
    public static void setHashField( String key, String field, String value) {
        List<String> list = (List<String>) toPayload(key, field, value);
        client.hset(list, res -> {
            if (res.succeeded()) {
                //System.out.println("hset result = " + res.result());
            } else {
                logger.error("redis error setHashField key:{} field:{} value:{} exception:{}", key, field, value, Tool.getException(res.cause()));
            }
        });
    }

    public static Future<Long> addHashField( String key, String field, long value) {
        Promise<Long> promise = Promise.promise();
        addHashField( key, field, value, promise);
        return promise.future();
    }

    public static void addHashField( String key, String field, long value, Handler<AsyncResult<Long>> onComplete) {
        client.hincrby(key, field, String.valueOf(value), res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toLong()));
            } else {
                logger.error("redis error addHashField key:{} field:{} value:{} exception:{}", key, field, value, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    public static void addHashFieldFloat( String key, String field, double value, Handler<AsyncResult<String>> onComplete) {
        client.hincrbyfloat(key, field, String.valueOf(value), res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toString()));
            } else {
                logger.error("redis error addHashFieldFloat key:{} field:{} value:{} exception:{}", key, field, value, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    public static void getHashField( String key, String field, Handler<AsyncResult<String>> onComplete) {
        client.hget(key, field, res -> {
            if (res.succeeded()) {
                if (res.result() == null) {
                    onComplete.handle(Future.succeededFuture(""));
                } else {
                    onComplete.handle(Future.succeededFuture(res.result().toString()));
                }
            } else {
                logger.error("redis error getHashField key:{} field:{} exception:{}", key, field, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    public static Future<String> getHashField( String key, String field) {
        Promise<String> promise = Promise.promise();
        getHashField( key, field, promise);
        return promise.future();
    }

    @SuppressWarnings("unchecked")
    public static void pushToList( String key, String value) {
        List<String> list = (List<String>) toPayload(key, value);
        client.rpush(list, res -> {
            if (res.succeeded()) {
                //System.out.println("rpush result = " + res.result());
            } else {
                logger.error("redis error pushToList key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
            }
        });
    }

    @SuppressWarnings("unchecked")
    public static void pushToListByLen( String key, String value, Handler<AsyncResult<Long>> onComplete) {
        List<String> list = (List<String>) toPayload(key, value);
        client.rpush(list, res -> {
            if (res.succeeded()) {
                //返回列表长度
                onComplete.handle(Future.succeededFuture(res.result().toLong()));
            } else {
                logger.error("redis error pushToListByLen key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    @SuppressWarnings("unchecked")
    public static void pushToList( String key, String value, Handler<AsyncResult<Boolean>> onComplete) {
        List<String> list = (List<String>) toPayload(key, value);
        client.rpush(list, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(true));
            } else {
                logger.error("redis error pushToList key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    @SuppressWarnings("unchecked")
    public static Future<Boolean> pushToList( String key, List<String> values) {
        Promise<Boolean> promise = Promise.promise();
        pushToList( key, values, promise);
        return promise.future();
    }

    @SuppressWarnings("unchecked")
    public static void pushToList( String key, List<String> values, Handler<AsyncResult<Boolean>> onComplete) {
        if (values.size() == 0) {
            onComplete.handle(Future.succeededFuture(false));
            return;
        } else {
            List<String> list = (List<String>) toPayload(key, values);
            client.rpush(list, res -> {
                if (res.succeeded()) {
                    onComplete.handle(Future.succeededFuture(true));
                } else {
                    logger.error("redis error pushToList key:{} values:{} exception:{}", key, list, Tool.getException(res.cause()));
                    onComplete.handle(Future.failedFuture(res.cause()));
                }
            });
        }
    }


    public static void pushToListByLen( String key, List<String> values, Handler<AsyncResult<Long>> onComplete) {
        if (values.size() == 0) {
            onComplete.handle(Future.succeededFuture(0L));
            return;
        } else {
            List<String> list = (List<String>) toPayload(key, values);
            client.rpush(list, res -> {
                if (res.succeeded()) {
                    onComplete.handle(Future.succeededFuture(res.result().toLong()));
                } else {
                    logger.error("redis error pushToList key:{} values:{} exception:{}", key, list, Tool.getException(res.cause()));
                    onComplete.handle(Future.failedFuture(res.cause()));
                }
            });
        }
    }

    /***
     * lsit指定区域保留，其他删除
     */
    public static void ltrim( String key, long from, long to) {
        client.ltrim(key, String.valueOf(from), String.valueOf(to), res -> {
            if (res.succeeded()) {
                //System.out.println("ltrim result = " + res.result());
            } else {
                logger.error("redis error ltrim key:{} from:{} to:{} exception:{}", key, from, to, Tool.getException(res.cause()));
            }
        });
    }

    public static void updateToList( String key, long index, String value) {
        client.lset(key, String.valueOf(index), value, res -> {
            if (res.succeeded()) {
                //System.out.println("lset result = " + res.result());
            } else {
                logger.error("redis error updateToList key:{} index:{} value:{} exception:{}", key, index, value, Tool.getException(res.cause()));
            }
        });
    }

    /**
     * 取列表中一段数据，例from=0,to=3,则取4个数据
     */
    public static void getListRange( String key, long from, long to, Handler<AsyncResult<JsonArray>> onComplete) {
        client.lrange(key, String.valueOf(from), String.valueOf(to), res -> {
            if (res.succeeded()) {
                JsonArray jsonArray = toJsonArray(res.result());
                onComplete.handle(Future.succeededFuture(jsonArray));
            } else {
                logger.error("redis error getListRange key:{} from:{} to:{} exception:{}", key, from, to, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /**
     * 获取某个hash obj的长度
     */
    public static void getHashLen( String key, Handler<AsyncResult<Long>> onComplete) {
        client.hlen(key, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toLong()));
            } else {
                logger.error("redis error getHashLen key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /**
     * 获取某个hash obj的长度
     */
    public static Future<Long> getHashLen( String key) {
        Promise<Long> promise = Promise.promise();
        getHashLen( key, promise);
        return promise.future();
    }

    /***
     * Redis Setnx（SET if Not eXists） 命令在指定的 key 不存在时，为 key 设置指定的值。如果key存在则不赋值 true表示有赋值，false表示没赋值
     */
    public static void setnx( String key, String value, Handler<AsyncResult<Boolean>> onComplete) {
        client.setnx(key, value, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toBoolean()));
            } else {
                logger.error("redis error setnx key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /***
     * Redis Setnx（SET if Not eXists） 命令在指定的 key 不存在时，为 key 设置指定的值。如果key存在则不赋值 true表示有赋值，false表示没赋值
     */
    public static Future<Boolean> setnx( String key, String value) {
        Promise<Boolean> promise = Promise.promise();
        setnx( key, value, promise);
        return promise.future();

    }

    /***
     * 获取key的过期时间
     */
    public static void ttl( String key, Handler<AsyncResult<Long>> onComplete) {
        client.ttl(key, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toLong()));
            } else {
                logger.error("redis error ttl key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }


    /***
     * 设置过期时间
     */
    public static void expire( String key, int seconds, Handler<AsyncResult<Boolean>> onComplete) {
        List<String> list = (List<String>) toPayload(key, String.valueOf(seconds));
        client.expire(list, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toBoolean()));
            } else {
                logger.error("redis error expire key:{} seconds:{} exception:{}", key, seconds, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /***
     * 设置过期时间
     */
    public static void expire( String key, int seconds) {
        List<String> list = (List<String>) toPayload(key, String.valueOf(seconds));
        client.expire(list, res -> {
            if (res.failed()) {
                logger.error("redis error expire key:{} seconds:{} exception:{}", key, seconds, Tool.getException(res.cause()));
            }
        });
    }

    @SuppressWarnings("unchecked")
    public static Future<Boolean> del( String key) {
        Promise<Boolean> promise = Promise.promise();
        del( key, promise);
        return promise.future();
    }

    @SuppressWarnings("unchecked")
    public static void del( String key, Handler<AsyncResult<Boolean>> onComplete) {
        List<String> list = (List<String>) toPayload(key);
        client.del(list, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toBoolean()));
            } else {
                logger.error("redis error del key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /**
     * 批量删除
     * 返回删除的数量
     */
    public static void delMany( List<String> keys, Handler<AsyncResult<Long>> onComplete) {
        if (keys.size() == 0) {
            onComplete.handle(Future.succeededFuture(0l));
            return;
        }
        client.del(keys, res -> {
            if (res.succeeded()) {
                long result = res.result().toLong();
                onComplete.handle(Future.succeededFuture(result));
            } else {
                logger.error("redis error delMany keys:{} exception:{}", keys, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }
    /**
     * 批量删除
     * 返回删除的数量
     */
    public static Future<Long> delMany( List<String> keys) {
        Promise<Long> promise = Promise.promise();
        delMany( keys, promise);
        return promise.future();
    }

//    /**
//     * 批量查询并删除（根据关键字前缀查询）
//     * 返回删除的数量
//     * 集群不支持scan，弃用
//     */
//    @Deprecated
//    public static void delMany( String prefix, int nodes, Handler<AsyncResult<Long>> onComplete) {
//        scan( prefix, nodes, res -> {
//            if (res.succeeded()) {
//                if (res.result() != null && res.result().size() > 0) {
//                    delMany( res.result().getList(), res2 -> {
//                        if (res2.succeeded()) {
//                            long result = res2.result();
//                            onComplete.handle(Future.succeededFuture(result));
//                        } else {
//                            logger.error("redis error delMany prefix:{} nodes:{} exception:{}", prefix, nodes, Tool.getException(res2.cause()));
//                            onComplete.handle(Future.failedFuture(res2.cause()));
//                        }
//                    });
//                }
//            } else {
//                logger.error("redis error delMany prefix:{} nodes:{} exception:{}", prefix, nodes, Tool.getException(res.cause()));
//                onComplete.handle(Future.failedFuture(res.cause()));
//            }
//        });
//    }

//    /**
//     * 批量查询（根据关键字前缀查询）
//     * 返回key列表
//     * 集群不支持keys，弃用
//     */
//    @Deprecated
//    public static void keys( String prefix, Handler<AsyncResult<JsonArray>> onComplete) {
//        prefix += "*";
//        String _prefix = prefix;
//        client.keys(prefix, res -> {
//            if (res.succeeded()) {
//                JsonArray result = toJsonArray(res.result());
//                onComplete.handle(Future.succeededFuture(result));
//            } else {
//                logger.error("redis error keys prefix:{} exception:{}", _prefix, Tool.getException(res.cause()));
//                onComplete.handle(Future.failedFuture(res.cause()));
//            }
//        });
//    }

//    /**
//     * 批量查询（根据关键字前缀查询）
//     *
//     * @param prefix 关键字前缀
//     * @param nodes  集群节点数量
//     *               返回key列表
//     *               集群不支持scan，弃用
//     */
//    @Deprecated
//    public static void scan( String prefix, int nodes, Handler<AsyncResult<JsonArray>> onComplete) {
//        prefix += "*";
//        List<Future> futures = new ArrayList<>();
//        for (int i = 0; i < nodes; i++) {
//            final int index = i;
//            Promise<JsonArray> future = Promise.promise();
//            futures.add(future.future());
//            List<String> list = (List<String>) toPayload(String.valueOf(i), new ScanOptions().setMatch(prefix).setCount(Integer.MAX_VALUE).toJsonArray());
//            client.scan(list, h -> {
//                if (h.succeeded()) {
//                    JsonArray result = toJsonArray(h.result()).getJsonArray(1);
//                    logger.info(" ### scan node " + index + " result size = " + result.size());
//                    future.complete(result);
//                } else {
//                    logger.error(" ### scan node " + index + " error !!!");
//                    logger.error(Tool.getException(h.cause()));
//                    future.complete(new JsonArray());
//                }
//            });
//        }
//        CompositeFuture.all(futures).onComplete(handler -> {
//            JsonArray result = new JsonArray();
//            for (Future fut : futures) {
//                JsonArray jsonArray = (JsonArray) fut.result();
//                result.addAll(jsonArray);
//            }
//            logger.info(" ### scan result size = " + result.size());
//            onComplete.handle(Future.succeededFuture(result));
//        });
//    }

    @SuppressWarnings("unchecked")
    public static void exists( String key, Handler<AsyncResult<Boolean>> onComplete) {
        List<String> list = (List<String>) toPayload(key);
        client.exists(list, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toBoolean()));
            } else {
                logger.error("redis error exists key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    @SuppressWarnings("unchecked")
    public static Future<Boolean> exists( String key) {
        Promise<Boolean> promise = Promise.promise();
        exists( key, promise);
        return promise.future();
    }

    public static void getset( String key, String value, Handler<AsyncResult<String>> onComplete) {
        client.getset(key, value, res -> {
            if (res.succeeded()) {
                Response response = res.result();
                onComplete.handle(Future.succeededFuture(response.toString()));
            } else {
                logger.error("redis error getset key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /**
     * 获取列表长度
     */
    public static void getLen( String key, Handler<AsyncResult<Long>> onComplete) {
        client.llen(key, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toLong()));
            } else {
                logger.error("redis error getLen key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /**
     * 通过索引获取列表中的元素
     */
    public static void getValueByIndex( String key, String index, Handler<AsyncResult<String>> onComplete) {
        client.lindex(key, index, res -> {
            if (res.succeeded()) {
                Response response = res.result();
                if (response == null) {
//                	logger.error("redis get response == null key = " + key);
                    onComplete.handle(Future.succeededFuture(null));
                } else {
                    onComplete.handle(Future.succeededFuture(response.toString()));
                }
            } else {
                logger.error("redis error getValueByIndex key:{} index:{} exception:{}", key, index, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /***
     * 移除表中所有与 VALUE 相等的值,并返回异步结果（列表为空时key会被自动移除）
     */
    public static void delToList( String key, String value, Handler<AsyncResult<Long>> onComplete) {
        client.lrem(key, "0", value, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toLong()));
            } else {
                logger.error("redis error delToList key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /***
     * 移除表中所有与 VALUE 相等的值,并返回异步结果（列表为空时key会被自动移除）
     */
    public static Future<Long> delToList( String key, String value) {
        Promise<Long> promise = Promise.promise();
        delToList( key, value, promise);
        return promise.future();
    }

    /**
     * 移除表中与 VALUE 相等的值, 数量为count,并返回异步结果（列表为空时key会被自动移除）
     *
     * @param key
     * @param value
     * @param count  从表头开始向表尾搜索，移除与 VALUE 相等的元素，数量为 COUNT. 从表尾开始向表头搜索，移除与 VALUE 相等的元素，数量为 COUNT 的绝对值
     */
    public static void delToList( String key, String value, String count) {
        client.lrem(key, count, value, res -> {
            if (res.failed()) {
                logger.error("redis error delToList key:{} value:{} count:{} exception:{}", key, value, count, Tool.getException(res.cause()));
            }
        });
    }

    // 没有则添加，有则覆盖值
    public static Future<Long> addRank( String key, long member, double score) {
        Promise<Long> promise = Promise.promise();
        addRank( key, member, score, promise);
        return promise.future();
    }

    // 没有则添加，有则覆盖值，返回最终结果值
    public static void addRank( String key, long member, double score, Handler<AsyncResult<Long>> onComplete) {
        List<String> list = (List<String>) toPayload(key, score, member);
        client.zadd(list, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toLong()));
            } else {
                logger.error("redis error addRank key:{} member:{} score:{} exception:{}", key, member, score, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    @SuppressWarnings("unchecked")
    // 分数前，ID后
    public static void addRankMany( String key, Map<String, Double> map) {
        if (map.isEmpty()) {
            return;
        }
        //此处zadd方法需要分数在前 key在后 所以这里写死map处理
        List<String> list = new ArrayList<>();
        list.add(key);
        for (Map.Entry<?, ?> pair : ((Map<?, ?>) map).entrySet()) {
            list.add(pair.getValue().toString());
            list.add(pair.getKey().toString());
        }
//        List<String> list = (List<String>) toPayload(key, map);

        client.zadd(list, res -> {
            if (res.succeeded()) {
                //System.out.println("zadd many result = " + res.result());
            } else {
                logger.error("redis error addRankMany key:{} list:{} exception:{}", key, list, Tool.getException(res.cause()));
            }
        });
    }

    /***
     * 获取榜单的长度
     */
    public static void getRankLen( String key, Handler<AsyncResult<Long>> onComplete) {
        client.zcard(key, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toLong()));
            } else {
                logger.error("redis error getRankLen key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /***
     * 获取榜单的长度
     */
    public static Future<Long> getRankLen( String key) {
        Promise<Long> promise = Promise.promise();
        getRankLen( key, promise);
        return promise.future();
    }

    @SuppressWarnings("unchecked")
    public static void removeFromRank( String key, long member) {
        List<String> list = (List<String>) toPayload(key, member);
        client.zrem(list, res -> {
            if (res.succeeded()) {
                //System.out.println("zrem result = " + res.result());
            } else {
                logger.error("redis error removeFromRank key:{} member:{} exception:{}", key, member, Tool.getException(res.cause()));
            }
        });
    }

    // 没有则添加，有则增量更新，返回最终结果值
    public static void updateRank( String key, long member, double score, Handler<AsyncResult<Long>> onComplete) {
        client.zincrby(key, String.valueOf(score), String.valueOf(member), res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toLong()));
            } else {
                logger.error("redis error updateRank key:{} member:{} score:{} exception:{}", key, member, score, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    // 没有则添加，有则增量更新，返回最终结果值
    public static Future<Double> updateRank( String key, long member, double score) {
        Promise<Double> fut = Promise.promise();
        client.zincrby(key, String.valueOf(score), String.valueOf(member), res -> {
            if (res.succeeded()) {
                //返回新增成员的数量
                fut.complete(Double.parseDouble(res.result().toString()));
            } else {
                logger.error("redis error updateRank key:{} member:{} score:{} exception:{}", key, member, score, Tool.getException(res.cause()));
                fut.fail(res.cause());
            }
        });
        return fut.future();
    }

//    /**
//     * 多个排行榜数据，合并到新排行榜 -- 集群不支持，弃用
//     * <p>多个排行榜并集，如果有重复member，score值将取和(ZUNIONSTORE命令默认设置)
//     *
//     * @param key        新排行榜key（如已存在，旧的将被覆盖）
//     * @param keyList    合并的排行榜key列表
//     * @param onComplete 返回新排行榜长度，执行失败时将返回-1
//     */
//    @Deprecated
//    public static void unionManyRanks( String key, List<String> keyList, Handler<AsyncResult<Long>> onComplete) {
//        List<String> list = (List<String>) toPayload(key, keyList.size(), keyList);
//        client.zunionstore(list, res -> {
//            if (res.succeeded()) {
//                onComplete.handle(Future.succeededFuture(res.result().toLong()));
//            } else {
//                logger.error("redis error unionManyRanks key:{} list:{} exception:{}", key, list, Tool.getException(res.cause()));
//                onComplete.handle(Future.failedFuture(res.cause()));
//            }
//        });
//    }

    /***
     * 根据排名区间获取排行列表(倒序，分数越大排越前面)
     * @param min 最小排名（从0开始）
     * @param max 最大排名
	 * @param withScore 查询结果是否带上分数（不带分数返回的是[id1,id2]，带分数返回的是[id1,score1,id2,score2]）
     */
    @SuppressWarnings("unchecked")
    public static void getRankListByIndex( String key, int min, int max, boolean withScore, Handler<AsyncResult<JsonArray>> onComplete) {
        List<String> list = (List<String>) toPayload(key, min, max);
        if (withScore) {
            list.add("WITHSCORES");
        }
        client.zrevrange(list, res -> {
            if (res.succeeded()) {
                JsonArray jsonArray = toJsonArray(res.result());
                onComplete.handle(Future.succeededFuture(jsonArray));
            } else {
                logger.error("redis error getRankListByIndex key:{} min:{} max:{} withScore:{} exception:{}", key, min, max, withScore, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /***
     * 根据排名区间获取排行列表(倒序，分数越大排越前面)
     * @param min 最小排名（从0开始）
     * @param max 最大排名
	 * @param withScore 查询结果是否带上分数（不带分数返回的是[id1,id2]，带分数返回的是[id1,score1,id2,score2]）
     */
    @SuppressWarnings("unchecked")
    public static Future<JsonArray> getRankListByIndex( String key, int min, int max, boolean withScore) {
        Promise<JsonArray> promise = Promise.promise();
        getRankListByIndex( key, min, max, withScore, promise);
        return promise.future();
    }

    /***
     * 根据排名区间获取排行列表(倒序，分数越大排越前面)
     * @param min 最小分数
     * @param max 最大分数
	 * @param withScore 查询结果是否带上分数（不带分数返回的是[id1,id2]，带分数返回的是[id1,score1,id2,score2]）
     */
    @SuppressWarnings("unchecked")
    public static void getRankListByScore( String key, long min, long max, boolean withScore, Handler<AsyncResult<JsonArray>> onComplete) {
        List<String> list = (List<String>) toPayload(key, max, min);
        if (withScore) {
            list.add("WITHSCORES");
        }
        client.zrevrangebyscore(list, res -> {
            if (res.succeeded()) {
                JsonArray jsonArray = toJsonArray(res.result());
                onComplete.handle(Future.succeededFuture(jsonArray));
            } else {
                logger.error("redis error getRankListByScore key:{} min:{} max:{} withScore:{} exception:{}", key, min, max, withScore, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

	/***
	 * 根据排名区间获取排行列表(倒序，分数越大排越前面)
	 * @param min 最小分数
	 * @param max 最大分数
	 * @param withScore 查询结果是否带上分数（不带分数返回的是[id1,id2]，带分数返回的是[id1,score1,id2,score2]）
	 */
	@SuppressWarnings("unchecked")
	public static Future<JsonArray> getRankListByScore(String key, long min, long max, boolean withScore) {
		Promise<JsonArray> promise = Promise.promise();
		getRankListByScore( key, min, max, withScore, promise);
		return promise.future();
	}

    /***
     * 获取我的排名(倒序，分数越大排越前面)
     */
    public static void getMyRank( String key, String id, Handler<AsyncResult<Long>> onComplete) {
        client.zrevrank(key, id, res -> {
            if (res.succeeded()) {
                long result = 0;
                Response response = res.result();
                if (response != null) {
                    result = response.toLong() + 1;// 排名是从0开始
                }
                onComplete.handle(Future.succeededFuture(result));
            } else {
                logger.error("redis error getMyRank key:{} id:{} exception:{}", key, id, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }
    /***
     * 获取我的排名(倒序，分数越大排越前面)
     */
    public static Future<Long> getMyRank( String key, String id) {
        Promise<Long> promise = Promise.promise();
        getMyRank( key, id, promise);
        return promise.future();

    }

    /***
     * 获取我的分数
     */
    public static void getMyScore( String key, long id, Handler<AsyncResult<Double>> onComplete) {
        client.zscore(key, String.valueOf(id), res -> {
            if (res.succeeded()) {
                double result = 0;
                Response response = res.result();
                if (response != null) {
                    result = response.toDouble();
                }
                onComplete.handle(Future.succeededFuture(result));
            } else {
                logger.error("redis error getMyScore key:{} id:{} exception:{}", key, id, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /***
     * 获取我的分数
     */
    public static Future<Double> getMyScore( String key, long id) {
        Promise<Double> promise = Promise.promise();
        getMyScore( key, id, promise);
        return promise.future();
    }

    /**
     * 往集合中添加元素，集合中的元素不可重复，无序
     */
    @SuppressWarnings("unchecked")
    public static void addToSet( String key, String value) {
        List<String> list = (List<String>) toPayload(key, value);
        client.sadd(list, res -> {
            if (res.succeeded()) {
                //System.out.println("sadd result = " + res.result());
            } else {
                logger.error("redis error addToSet key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
            }
        });
    }


    /**
     * 往集合中添加元素，集合中的元素不可重复，无序
     */
    @SuppressWarnings("unchecked")
    public static void addToSet( String key, List<String> value) {
        List<String> list = (List<String>) toPayload(key, value);
        client.sadd(list, res -> {
            if (res.succeeded()) {
                //System.out.println("sadd result = " + res.result());
            } else {
                logger.error("redis error addToSet key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
            }
        });
    }

    /**
     * 往集合中添加元素，集合中的元素不可重复，无序
     */
    @SuppressWarnings("unchecked")
    public static void addToSet( String key, String value, Handler<AsyncResult<Integer>> onComplete) {
        List<String> list = (List<String>) toPayload(key, value);
        client.sadd(list, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toInteger()));
            } else {
                logger.error("redis error addToSet key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /**
     * 删除集合中的元素
     */
    @SuppressWarnings("unchecked")
    public static void delToSet( String key, String value) {
        List<String> list = (List<String>) toPayload(key, value);
        client.srem(list, res -> {
            if (res.succeeded()) {
                //System.out.println("removeToSet result = " + res.result());
            } else {
                logger.error("redis error delToSet key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
            }
        });
    }

    /**
     * 删除集合中的元素
     */
    @SuppressWarnings("unchecked")
    public static void delToSet( String key, String value, Handler<AsyncResult<Integer>> onComplete) {
        List<String> list = (List<String>) toPayload(key, value);
        client.srem(list, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toInteger()));
            } else {
                logger.error("redis error delToSet key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /**
     * 获取集合所有元素
     */
    public static void getFullSet( String key, Handler<AsyncResult<JsonArray>> onComplete) {
        client.smembers(key, res -> {
            if (res.succeeded()) {
                JsonArray jsonArray = toJsonArray(res.result());
                onComplete.handle(Future.succeededFuture(jsonArray));
            } else {
                logger.error("redis error getFullSet key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /**
     * 获取set长度
     */
    public static void getSetLen( String key, Handler<AsyncResult<Long>> onComplete) {
        client.scard(key, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toLong()));
            } else {
                logger.error("redis error getSetLen key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /**
     * 判断value是否set成员
     */
    public static void sismember( String key, String value, Handler<AsyncResult<Boolean>> onComplete) {
        client.sismember(key, value, res -> {
            if (res.succeeded()) {
                int result = res.result().toInteger();
                onComplete.handle(Future.succeededFuture(result == 1));
            } else {
                logger.error("redis error sismember key:{} value:{} exception:{}", key, value, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /***
     * 获取列表的全部内容
     * @param key
     * @param onComplete
     */
    public static void getJsonArray( String key, Handler<AsyncResult<JsonArray>> onComplete) {
        client.lrange(key, "0", "-1", res -> {
            if (res.failed()) {
                logger.error("redis error getJsonArray key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            } else {
                JsonArray result = toJsonArray(res.result());
                onComplete.handle(Future.succeededFuture(result));
            }
        });
    }

    public static void getJsonArray( String key, long from, long to, Handler<AsyncResult<JsonArray>> onComplete) {
        client.lrange(key, String.valueOf(from), String.valueOf(to), res -> {
            if (res.failed()) {
                logger.error("redis error getJsonArray key:{} from:{} to:{} exception:{}", key, from, to, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            } else {
                JsonArray result = toJsonArray(res.result());
                onComplete.handle(Future.succeededFuture(result));
            }
        });
    }

    public static Future<JsonArray> getJsonArray( String key) {
        Promise<JsonArray> promise = Promise.promise();
        getJsonArray( key, promise);
        return promise.future();
    }

    /**
     * 删除列表最早的值
     *
     * @return 被删除的记录
     */
    public static void lpop( String key, Handler<AsyncResult<String>> onComplete) {
        List<String> list = new ArrayList<>();
        list.add(key);
        client.lpop(list, res -> {
            if (res.failed()) {
                logger.error("redis error lpop key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            } else {
                String result;
                if (res.result() == null) {
                    result = "";
                } else {
                    result = res.result().toString();
                }
                onComplete.handle(Future.succeededFuture(result));
            }
        });
    }
    /**
     * 删除列表最早的值
     *
     * @return 被删除的记录
     */
    public static Future<String> lpop( String key) {
        Promise<String> promise = Promise.promise();
        lpop( key, promise);
        return promise.future();
    }

    /**
     * 删除列表最新的值
     *
     * @return 被删除的记录
     */
    public static void rpop( String key, Handler<AsyncResult<String>> onComplete) {
        List<String> list = new ArrayList<>();
        list.add(key);
        client.rpop(list, res -> {
            if (res.failed()) {
                logger.error("redis error rpop key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            } else {
                String result;
                if (res.result() == null) {
                    result = "";
                } else {
                    result = res.result().toString();
                }
                onComplete.handle(Future.succeededFuture(result));
            }
        });
    }

    /***
     * 增量更新排行榜,并获取更新后我的排名
     * @param key
     * @param member
     * @param score
     * @param onComplete
     */
    public static void updateRankAndGetMyRank( String key, long member, double score, Handler<AsyncResult<Long>> onComplete) {
        client.zincrby(key, String.valueOf(score), "" + member, res -> {
            if (res.succeeded()) {
                //返回新增成员的数量
                getMyRank( key, "" + member, onComplete);
            } else {
                logger.error("redis error updateRankAndGetMyRank key:{} member:{} score:{} exception:{}", key, member, score, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });

    }

    /***
     * 往列表头添加一条数据然后删除列表的尾部的数据，只能是string型，一般是JsonObject转成string存入
     * @param key
     * @param value
     * @param num 列表保留的长度
     */
    public static void pushToListAndTrim( String key, String value, int num) {
        List<String> list = (List<String>) toPayload(key, value);
        client.lpush(list, res -> {
            if (res.failed()) {
                logger.error("redis error pushToListAndTrim key:{} value:{} num:{} exception:{}", key, value, num, Tool.getException(res.cause()));
            } else {
                ltrim( key, 0, num - 1);
            }
        });
    }

    /**
     * 给hash字段复制，设置成功，返回 1 。 如果给定字段已经存在且没有操作被执行，返回 0
     */
    public static void hsetnx( String key, String field, String value, Handler<AsyncResult<Boolean>> onComplete) {
        client.hsetnx(key, field, value, res -> {
            if (res.succeeded()) {
                int result = res.result().toInteger();
                onComplete.handle(Future.succeededFuture(result == 1));
            } else {
                logger.error("redis error hsetnx key:{} field:{} value:{} exception:{}", key, field, value, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    public static Future<Boolean> hsetnx( String key, String field, String value) {
        Promise<Boolean> promise = Promise.promise();
        hsetnx( key, field, value, promise);
        return promise.future();
    }

    // 删除一个hash key
    public static void hdel( String key, String field, Handler<AsyncResult<Boolean>> onComplete) {
        List<String> list = (List<String>) toPayload(key, field);
        client.hdel(list, res -> {
            if (res.succeeded()) {
                int result = res.result().toInteger();
                onComplete.handle(Future.succeededFuture(result > 0));
            } else {
                logger.error("redis error hdel key:{} field:{} exception:{}", key, field, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    public static Future<Boolean> hdel( String key, String field) {
        Promise<Boolean> promise = Promise.promise();
        hdel( key, field, promise);
        return promise.future();
    }

    // 删除多个hash key
    public static void hdel( String key, JsonArray jsonArray, Handler<AsyncResult<Boolean>> onComplete) {
        List<String> list = (List<String>) toPayload(key, jsonArray);
        client.hdel(list, res -> {
            if (res.succeeded()) {
                int result = res.result().toInteger();
                onComplete.handle(Future.succeededFuture(result > 0));
            } else {
                logger.error("redis error hdel key:{} fields:{} exception:{}", key, jsonArray, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    //查看hash 指定字段是否存在。
    public static void hexists( String rediskey, String key, Handler<AsyncResult<Boolean>> onComplete) {
        client.hexists(rediskey, key, res -> {
            if (res.succeeded()) {
                Boolean result = res.result().toInteger() == 1 ? true : false;
                onComplete.handle(Future.succeededFuture(result));
            } else {
                logger.error("redis error hexists key:{} field:{} exception:{}", rediskey, key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    /***
     * 获取hash对象所有key
     */
    public static void getHashKeys( String key, Handler<AsyncResult<JsonArray>> onComplete) {
        client.hkeys(key, res -> {
            if (res.failed()) {
                logger.error("redis error getHashKeys key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            } else {
                JsonArray result = toJsonArray(res.result());
                onComplete.handle(Future.succeededFuture(result));
            }
        });
    }

    public static Future<JsonArray> getHashKeys( String key) {
        Promise<JsonArray> promise = Promise.promise();
        getHashKeys( key, promise);
        return promise.future();
    }

    /***
     * 给key赋值并且设置过期时间
     */
    public static void setAndExpire( String key, String value, int seconds, Handler<AsyncResult<Boolean>> onComplete) {
        client.setex(key, String.valueOf(seconds), value, res -> {
            if (res.succeeded()) {
				onComplete.handle(Future.succeededFuture(res.result().equals("OK")));
			} else {
                logger.error("redis error setAndExpire key:{} value:{} seconds:{} exception:{}", key, value, seconds, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

    public static void setAndExpire( String key, String value, int seconds) {
        client.setex(key, String.valueOf(seconds), value, res -> {
            if (res.failed()) {
                logger.error("redis error setAndExpire key:{} value:{} seconds:{} exception:{}", key, value, seconds, Tool.getException(res.cause()));
            }
        });
    }

    /**
     * Redis key类型名称
     */
    public enum RedisKeyTypeEnum {
        NONE("none"),// key不存在
        STRING("string"),//字符串
        HASH("hash"),// 哈希（一般对象数据在用）
        LIST("list"),// 列表
        SET("set"),//集合
        ZSET("zset"),//有序集合
        STREAM("stream");//流

        public final String value;

        RedisKeyTypeEnum(String val) {
            value = val;
        }
    }

    /**
     * 获取key的类型
     *
     * @param onComplete 返回key类型（可用{@link RedisKeyTypeEnum}），不存在返回none
     */
    public static void type( String key, Handler<AsyncResult<String>> onComplete) {
        client.type(key, res -> {
            if (res.succeeded()) {
                onComplete.handle(Future.succeededFuture(res.result().toString()));
            } else {
                logger.error("redis error type key:{} exception:{}", key, Tool.getException(res.cause()));
                onComplete.handle(Future.failedFuture(res.cause()));
            }
        });
    }

}
